package net.bwie.realtime.vehicle2.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.vehicle.bean.VehicleData;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Author: FuHe
 * @Date: 2025/6/8
 */
public class a {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //  读取数据 kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle-data-log");
//        kafkaStream.print();
        // 处理数据
        DataStream<String> processStream = handle(kafkaStream);
        processStream.print();
        // 写入数据 clickhouse
        JdbcUtil.sinkToClickhouseUpsert(processStream,
                "INSERT INTO vehicle_log.vehicle_abnormal_alarm (window_start, window_end, vin, alarm_type) " +
                        "VALUES (?, ?, ?, ?)");
        env.execute("VehicleAbnormalAlarmJob2");
    }

    private static DataStream<String> handle(DataStream<String> kafkaStream) {
        SingleOutputStreamOperator<net.bwie.realtime.vehicle2.bean.VehicleData> map = kafkaStream.map(o -> JSON.parseObject(o, net.bwie.realtime.vehicle2.bean.VehicleData.class));
        SingleOutputStreamOperator<net.bwie.realtime.vehicle2.bean.VehicleData> ts = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<net.bwie.realtime.vehicle2.bean.VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<net.bwie.realtime.vehicle2.bean.VehicleData>() {
                            @Override
                            public long extractTimestamp(net.bwie.realtime.vehicle2.bean.VehicleData element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }
                )
        );
        KeyedStream<net.bwie.realtime.vehicle2.bean.VehicleData, String> vinStream = ts.keyBy(o -> o.getVin());

        // 1. 检测低电量异常(1分钟内出现3次低于10%)
        Pattern<VehicleData, ?> batteryLevelPattern = Pattern.<VehicleData>begin("batteryLevel")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getBatteryLevel() != null && value.getBatteryLevel() < 10.0;
                    }
                })
                .times(3)
                .within(Time.minutes(1));
        // 2. 检测高温异常(电池温度连续3次高于60度)
        Pattern<VehicleData, ?> batteryTempPattern = Pattern.<VehicleData>begin("batteryTemp")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getBatteryTemp() != null && value.getBatteryTemp() > 60.0;
                    }
                })
                .times(3)
                .within(Time.minutes(1));
        // 3. 检测高温异常(电机温度连续3次高于60度)
        Pattern<VehicleData, ?> motorTempPattern = Pattern.<VehicleData>begin("motorTemp")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getMotorTemp() != null && value.getMotorTemp() > 60.0;
                    }
                })
                .times(3)
                .within(Time.minutes(1));
        // 4. 速度异常(速度连续3次超过120)
        Pattern<VehicleData, ?> speedPattern = Pattern.<VehicleData>begin("speed")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getSpeed() != null && value.getSpeed() > 120;
                    }
                })
                .times(3)
                .within(Time.minutes(1));

//        PatternStream<VehicleData> batteryLevelStream = CEP.pattern(
//                vehicleStream.keyBy(VehicleData::getVin),
//                batteryLevelPattern);
//        PatternStream<VehicleData> batteryTempStream = CEP.pattern(
//                vehicleStream.keyBy(VehicleData::getVin),
//                batteryTempPattern);

        // 处理检测到的模式
//        SingleOutputStreamOperator<String> lowBatteryAlarms = batteryLevelStream.process(
//                new PatternProcessFunction<VehicleData, String>() {
//                    @Override
//                    public void processMatch(Map<String, List<VehicleData>> match, Context ctx, Collector<String> out) {
//                        VehicleData firstEvent = match.get("firstLowBattery").get(0);
//                        long windowStart = firstEvent.getTimestamp() - (firstEvent.getTimestamp() % 60000);
//                        long windowEnd = windowStart + 60000;

//                        out.collect(String.format("%d,%d,%s,电池电量低于10%%",
//                                windowStart, windowEnd, firstEvent.getVin()));
//                        out.collect(windowStart+","+windowEnd,firstEvent.getVin(),"电池电量低于10%%");
//                    }
//                });
//
//        SingleOutputStreamOperator<String> highTempAlarms = batteryTempStream.process(
//                new PatternProcessFunction<VehicleData, String>() {
//                    @Override
//                    public void processMatch(
//                            Map<String, List<VehicleData>> match,
//                            Context ctx,
//                            Collector<String> out) {
//
//                        VehicleData firstEvent = match.get("firstHighTemp").get(0);
//                        out.collect(String.format("%d,%d,%s,电池温度高于60度",
//                                firstEvent.getTimestamp(),
//                                firstEvent.getTimestamp(),
//                                firstEvent.getVin()));
//                    }
//                });

        // 合并所有报警流
//        return lowBatteryAlarms.union(highTempAlarms);
        return null;
    }
}
